#include "inspector.h"

#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "config.h"
#include "redismodule.h"

#define SYSTEM_ID_LEN 4
#define SAMPLE_KEY_SIZE 10

void inspectorTimerHandler(RedisModuleCtx *ctx, void *data);

static const char *config_path = NULL;

static TaskConfig task_config;
static TaskInfo task_info;
static dict *shadow_sys_data;
static m_zskiplist *shadow_big_keys;

const char *interanl_keys[]
    = {"__exporter_", "cluster_proxy_set", "cluster_proxy_statistics", "cluster_proxy_for_keepalive"};

void _moduleAssert(const char *estr, const char *file, int line) {
    fprintf(stderr, "=== ASSERTION FAILED ===");
    fprintf(stderr, "==> %s:%d '%s' is not true", file, line, estr);
    *((char *)-1) = 'x';
}

uint64_t dictModuleStrHash(const void *key) {
    size_t len;
    const char *buf = RedisModule_StringPtrLen(key, &len);
    return m_dictGenHashFunction(buf, (int)len);
}

int dictModuleStrKeyCompare(void *privdata, const void *key1, const void *key2) {
    size_t l1, l2;
    DICT_NOTUSED(privdata);

    const char *buf1 = RedisModule_StringPtrLen(key1, &l1);
    const char *buf2 = RedisModule_StringPtrLen(key2, &l2);
    if (l1 != l2) return 0;
    return memcmp(buf1, buf2, l1) == 0;
}

void dictModuleKeyDestructor(void *privdata, void *key) {
    DICT_NOTUSED(privdata);
    if (key) {
        RedisModule_FreeString(NULL, key);
        key = NULL;
    }
}

void dictModuleValueDestructor(void *privdata, void *val) {
    DICT_NOTUSED(privdata);
    if (val) {
        SystemUsage *v = (SystemUsage *)val;
        if (v->big_key_sample) m_zslFree(v->big_key_sample);
        if (v->ttl_key_sample) m_listEmpty(v->ttl_key_sample);
        zfree(val);
        val = NULL;
    }
}

m_dictType inspectDataDictType = {
    dictModuleStrHash,        /* hash function */
    NULL,                     /* key dup */
    NULL,                     /* val dup */
    dictModuleStrKeyCompare,  /* key compare */
    dictModuleKeyDestructor,  /* key destructor */
    dictModuleValueDestructor /* val destructor */
};

void listFreeElement(void *ptr) {
    if (ptr) {
        RedisModule_FreeString(NULL, ptr);
        ptr = NULL;
    }
}

// 相等返回0
static int mstrcasecmp(const RedisModuleString *rs1, const char *s2) {
    size_t n1 = strlen(s2);
    size_t n2;
    const char *s1 = RedisModule_StringPtrLen(rs1, &n2);
    if (n1 != n2) {
        return -1;
    }
    return strncasecmp(s1, s2, n1);
}

void reset_task_info() {
    task_info.is_running = 0;
    task_info.start_time = 0;
    task_info.cost_time = 0;
    task_info.count = 0;
    task_info.dbsize = 0;
    if (task_info.sys_data) m_dictEmpty(task_info.sys_data, NULL);
    if (task_info.global_big_key) m_zslDeleteRangeByRank(task_info.global_big_key, 0, task_info.global_big_key->length);
    if (shadow_sys_data) {
        m_dictRelease(shadow_sys_data);
        shadow_sys_data = NULL;
    }
    if (shadow_big_keys) {
        m_zslFree(shadow_big_keys);
        shadow_big_keys = NULL;
    }
}

/**
 * inspector module提供的相关命令
 */

// format: inspector.config get key
//         inspector.config set key value
// eg:     inspector config set enable 1/0
// ---
// key: enable, interval, scan_tick, scan_count,
//      items_threshold,  memory_threshold, ttl_threshold
int InspectorConfig_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    RedisModule_AutoMemory(ctx);

    // syntax check
    if (argc < 3) {
        return RedisModule_WrongArity(ctx);
    } else if (mstrcasecmp(argv[1], "get") && mstrcasecmp(argv[1], "set")) {
        RedisModule_ReplyWithError(ctx, "ERR Invalid config cmd");
        return REDISMODULE_OK;
    }

    long long value = 0;
    int is_set = 0;
    if (!mstrcasecmp(argv[1], "set")) {
        if (argc < 4) {
            return RedisModule_WrongArity(ctx);
        } else if (RedisModule_StringToLongLong(argv[3], &value) != REDISMODULE_OK) {
            RedisModule_ReplyWithError(ctx, "ERR Parse config value fail");
            return REDISMODULE_OK;
        }
        is_set = 1;
        RedisModule_Log(ctx, "notice", "set config %s to %s", RedisModule_StringPtrLen(argv[2], NULL),
                        RedisModule_StringPtrLen(argv[3], NULL));
    } else {
        RedisModule_Log(ctx, "notice", "get config %s", RedisModule_StringPtrLen(argv[2], NULL));
    }

    int is_64bit = 0;
    uintptr_t *old_ptr;
    if (!mstrcasecmp(argv[2], "enable")) {
        old_ptr = (uintptr_t *)&(task_config.enable);
        // disable时清空状态信息
        if (is_set && value == 0) {
            reset_task_info();
        }
    } else if (!mstrcasecmp(argv[2], "interval")) {
        old_ptr = (uintptr_t *)&(task_config.interval);
    } else if (!mstrcasecmp(argv[2], "exclusive")) {
        old_ptr = (uintptr_t *)&(task_config.exclusive);
    } else if (!mstrcasecmp(argv[2], "scan_tick")) {
        old_ptr = (uintptr_t *)&(task_config.scan_tick);
    } else if (!mstrcasecmp(argv[2], "scan_count")) {
        old_ptr = (uintptr_t *)&(task_config.scan_count);
    } else if (!mstrcasecmp(argv[2], "items_threshold")) {
        old_ptr = (uintptr_t *)&(task_config.items_threshold);
    } else if (!mstrcasecmp(argv[2], "memory_threshold")) {
        is_64bit = 1;
        old_ptr = (uintptr_t *)(&task_config.memory_threshold);
    } else if (!mstrcasecmp(argv[2], "ttl_threshold")) {
        old_ptr = (uintptr_t *)&(task_config.ttl_threshold);
    } else {
        RedisModule_ReplyWithError(ctx, "ERR Config key not support");
        return REDISMODULE_OK;
    }

    if (is_set) {
        if (is_64bit) {
            *(uint64_t *)old_ptr = (uint64_t)value;
        } else {
            *(uint32_t *)old_ptr = (uint32_t)value;
        }
        RedisModule_ReplyWithSimpleString(ctx, "OK");
        if (config_path != NULL) {
            flush_config(ctx, config_path, &task_config);
        }
    } else {
        RedisModule_ReplyWithLongLong(ctx, is_64bit ? *(uint64_t *)old_ptr : *(int32_t *)old_ptr);
    }

    return REDISMODULE_OK;
}

int InspectorInfo_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    REDISMODULE_NOT_USED(argv);
    REDISMODULE_NOT_USED(argc);

    RedisModule_AutoMemory(ctx);

    uint64_t scan_capacity = 0;
    RedisModuleString *info_usage = RedisModule_CreateStringPrintf(ctx, "\r\n# System Info\r\n");

    m_dictEntry *entry;
    m_dictIterator *it = m_dictGetIterator(task_info.sys_data);
    while ((entry = m_dictNext(it))) {
        RedisModuleString *key = (RedisModuleString *)dictGetKey(entry);
        const char *system_id = RedisModule_StringPtrLen(key, NULL);
        SystemUsage *sys_usage = (SystemUsage *)dictGetVal(entry);
        char buf[512] = {0};
        int len = sprintf(buf,
                          "%s: capacity:%lu,key_cnt:%u,big_key_cnt:%u,big_key_mem:%lu,ttl_key_cnt:%u,ttl_key_mem:%lu,"
                          "k_string:%u,k_list:%u,k_hash:%u,k_set:%u,k_zset:%u,k_tair_string:%u,k_tair_hash:%u,"
                          "k_other:%u\r\n",
                          system_id, sys_usage->capacity, sys_usage->key_cnt, sys_usage->big_key_cnt,
                          sys_usage->big_key_capacity, sys_usage->ttl_key_cnt, sys_usage->ttl_key_capacity,
                          sys_usage->key_cnt_string, sys_usage->key_cnt_list, sys_usage->key_cnt_hash,
                          sys_usage->key_cnt_set, sys_usage->key_cnt_zset, sys_usage->key_cnt_tair_string,
                          sys_usage->key_cnt_tair_hash, sys_usage->key_cnt_other);
        RedisModule_StringAppendBuffer(ctx, info_usage, buf, len);
        scan_capacity += sys_usage->capacity;
    }
    m_dictReleaseIterator(it);

    // 格式化时间戳
    time_t seconds = task_info.start_time / 1000;
    long millis = task_info.start_time % 1000;
    struct tm *ts;
    ts = localtime((const time_t *)&seconds);
    char ts_str[80] = {0};
    strftime(ts_str, sizeof(ts_str), "%Y-%m-%d %H:%M:%S", ts);

    // 计算进度百分比
    float progress = task_info.dbsize != 0 ? ((float)task_info.count / task_info.dbsize) : 0.0;

    RedisModuleString *info_status = RedisModule_CreateStringPrintf(
        ctx,
        "# Status\r\n"
        "running:%s\r\n"
        "exclusive:%s\r\n"
        "start_time:%s.%03ld\r\n"
        "cost_time:%ld ms\r\n"
        "progress:%2.2f%% (%u / %ld)\r\n"
        "scan_capacity:%ld\r\n",  // 总共扫描的key容量
        task_info.is_running ? "yes" : "no", task_config.exclusive ? "yes" : "no", ts_str, millis, task_info.cost_time,
        progress * 100, task_info.count, task_info.dbsize, scan_capacity);

    size_t buf_size = 0;
    char buf[1024 * 1024] = {0};

    size_t len = 0;
    const char *status_buf = RedisModule_StringPtrLen(info_status, &len);
    strncat(buf, status_buf, len);
    buf_size += len;

    const char *usage_buf = RedisModule_StringPtrLen(info_usage, &len);
    strncat(buf, usage_buf, len);
    buf_size += len;

    RedisModule_ReplyWithStringBuffer(ctx, buf, buf_size);

    return REDISMODULE_OK;
}

// eg: inspector.keys big_key/ttl_key system_id
int InspectorKeys_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    RedisModule_AutoMemory(ctx);
    if (argc < 3) {
        return RedisModule_WrongArity(ctx);
    }

    if (mstrcasecmp(argv[1], "TTL_KEY") && mstrcasecmp(argv[1], "BIG_KEY")) {
        RedisModule_ReplyWithError(ctx, "ERR Type must be ttl_key or big_key");
        return REDISMODULE_OK;
    }

    if (!mstrcasecmp(argv[1], "BIG_KEY")) {
        if (!mstrcasecmp(argv[2], "ALL")) {
            // 查询全局级别的big_key
            unsigned long len = task_info.global_big_key->length;
            if (len == 0) {
                RedisModule_ReplyWithEmptyArray(ctx);
                return REDISMODULE_OK;
            } else {
                RedisModule_ReplyWithArray(ctx, len * 2);
                for (size_t i = 1; i <= len; i++) {
                    m_zskiplistNode *node = m_zslGetElementByRank(task_info.global_big_key, i);
                    RedisModule_ReplyWithString(ctx, node->member);
                    RedisModule_ReplyWithString(ctx, RedisModule_CreateStringFromDouble(ctx, node->score));
                }
            }
        } else {
            // 查询子系统级别的big_key
            SystemUsage *sys_usage = (SystemUsage *)m_dictFetchValue(task_info.sys_data, argv[2]);
            if (sys_usage == NULL) {
                RedisModule_ReplyWithEmptyArray(ctx);
                return REDISMODULE_OK;
            }
            size_t len = sys_usage->big_key_sample->length;
            if (len == 0) {
                RedisModule_ReplyWithEmptyArray(ctx);
                return REDISMODULE_OK;
            }
            RedisModule_ReplyWithArray(ctx, len);
            for (size_t i = 1; i <= len; i++) {
                m_zskiplistNode *node = m_zslGetElementByRank(sys_usage->big_key_sample, i);
                RedisModule_ReplyWithString(ctx, node->member);
            }
        }
    } else {
        // 查询子系统级别ttl_key
        SystemUsage *sys_usage = (SystemUsage *)m_dictFetchValue(task_info.sys_data, argv[2]);
        if (sys_usage == NULL) {
            RedisModule_ReplyWithEmptyArray(ctx);
            return REDISMODULE_OK;
        }
        size_t len = listLength(sys_usage->ttl_key_sample);
        if (len == 0) {
            RedisModule_ReplyWithEmptyArray(ctx);
            return REDISMODULE_OK;
        }

        RedisModule_ReplyWithArray(ctx, len);
        m_listNode *node;
        m_listIter *it = m_listGetIterator(sys_usage->ttl_key_sample, AL_START_HEAD);
        while ((node = m_listNext(it))) {
            RedisModule_ReplyWithString(ctx, listNodeValue(node));
        }
        m_listReleaseIterator(it);
    }

    return REDISMODULE_OK;
}

// 立即触发扫描, 若有正在执行的扫描任务, 则返回失败
// format: inspector.trigger
int InspectorTrigger_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    REDISMODULE_NOT_USED(argv);
    REDISMODULE_NOT_USED(argc);

    RedisModule_AutoMemory(ctx);

    if (!task_config.enable) {
        RedisModule_ReplyWithError(ctx, "ERR Can not trigger scan when inspector module disabled");
        return REDISMODULE_OK;
    } else if (RedisModule_GetContextFlags(ctx) & REDISMODULE_CTX_FLAGS_MASTER) {
        RedisModule_ReplyWithError(ctx, "ERR Can not trigger scan task on master");
        return REDISMODULE_OK;
    } else if (task_info.is_running) {
        RedisModule_ReplyWithError(ctx, "ERR Inspector scan task is running");
        return REDISMODULE_OK;
    }

    RedisModule_Log(ctx, "notice", "manual triger a inspector scan task");
    // 让timer能发起新一轮扫描
    task_info.start_time = 0;
    task_info.cost_time = 0;

    RedisModule_ReplyWithSimpleString(ctx, "OK");
    return REDISMODULE_OK;
}

// 独占模式下, 不识别子系统id, 统一归为9999
RedisModuleString *extract_sys_id(RedisModuleCtx *ctx, RedisModuleString *key) {
    char prefix[4] = "9999";
    if (!task_config.exclusive) {
        size_t len;
        const char *ptr = RedisModule_StringPtrLen(key, &len);
        if (len >= 4) {
            strncpy(prefix, ptr, 4);
        }
    }
    return RedisModule_CreateString(ctx, prefix, 4);
}

// 返回0表示不是big key, -1表示在检测时key已不存在
int is_big_key(RedisModuleCtx *ctx, RedisModuleString *key, RedisModuleKey *real_key, uint64_t *mem, int *key_type) {
    int type = RedisModule_KeyType(real_key);
    *key_type = type;
    if (type == REDISMODULE_KEYTYPE_STRING || type == REDISMODULE_KEYTYPE_LIST || type == REDISMODULE_KEYTYPE_HASH
        || type == REDISMODULE_KEYTYPE_SET || type == REDISMODULE_KEYTYPE_ZSET || type == REDISMODULE_KEYTYPE_MODULE) {
        // 若为集合类型且非moudle自定义类型, 则需获取元素个数
        size_t items_len = 1;
        if (type != REDISMODULE_KEYTYPE_STRING && type != REDISMODULE_KEYTYPE_MODULE) {
            items_len = RedisModule_ValueLength(real_key);
        }

        // 若为module类型, 则需调用type命令获取实际类型
        if (type == REDISMODULE_KEYTYPE_MODULE) {
            RedisModuleCallReply *type_reply = RedisModule_Call(ctx, "TYPE", "s", key);
            if (type_reply == NULL || RedisModule_CallReplyType(type_reply) != REDISMODULE_REPLY_STRING) {
                return -1;
            }

            RedisModuleString *type_str = RedisModule_CreateStringFromCallReply(type_reply);
            if (!mstrcasecmp(type_str, "exstrtype")) {
                *key_type = REDISMODULE_KEYTYPE_MODULE_TAIR_STRING;
            } else if (!mstrcasecmp(type_str, "tairhash-")) {
                *key_type = REDISMODULE_KEYTYPE_MODULE_TAIR_HASH;
                // TODO:
                // 单独处理tairhash的memory usage和items len
                /* RedisModule_Call(ctx, "EXHLEN", "sc", key, "noExp");
                RedisModule_Call(ctx, "DUMP", "s", key); */
                return 0;
            }
        }

        RedisModuleCallReply *mem_reply = RedisModule_Call(ctx, "MEMORY", "cscl", "USAGE", key, "SAMPLES", "0");
        if (mem_reply == NULL || RedisModule_CallReplyType(mem_reply) != REDISMODULE_REPLY_INTEGER) {
            return -1;
        }
        RedisModule_StringToLongLong(RedisModule_CreateStringFromCallReply(mem_reply), (long long *)mem);
        return items_len > task_config.items_threshold || *mem > task_config.memory_threshold;
    } else if (type == REDISMODULE_KEYTYPE_EMPTY) {
        return -1;
    } else {
        RedisModule_Log(ctx, "warning", "unsupport type on check big key");
        return -1;
    }
}

int is_ttl_key(RedisModuleKey *real_key) {
    mstime_t expire = RedisModule_GetExpire(real_key);
    if (expire == -2) return -1;
    return expire == -1 || expire > task_config.ttl_threshold;
}

// 跳过内部使用的key
int skip_internal_key(RedisModuleString *key) {
    size_t len;
    const char *key_name = RedisModule_StringPtrLen(key, &len);
    size_t arr_len = sizeof(interanl_keys) / sizeof(interanl_keys[0]);
    for (size_t i = 0; i < arr_len; i++) {
        if (!strncasecmp(key_name, interanl_keys[i], sizeof(*interanl_keys[i]))) {
            return 1;
        }
    }
    return 0;
}

/**
 * 定时巡检内存逻辑
 */
void inspectorTimerHandler(RedisModuleCtx *ctx, void *data) {
    REDISMODULE_NOT_USED(data);
    RedisModule_AutoMemory(ctx);

    // 只在从节点开启扫描
    if (!task_config.enable || (RedisModule_GetContextFlags(ctx) & REDISMODULE_CTX_FLAGS_MASTER)) {
        reset_task_info();
        goto restart;
    }

    if (!task_info.is_running) {
        // 判断距上一轮扫描时间
        long long now = RedisModule_Milliseconds();
        if (now < task_info.start_time + task_info.cost_time + task_config.interval) {
            goto restart;
        }
        // 开启新一轮扫描
        task_info.is_running = 1;
        task_info.start_time = now;
        task_info.cost_time = 0;
        task_info.count = 0;
        task_info.dbsize = RedisModule_DbSize(ctx);
        shadow_sys_data = m_dictCreate(&inspectDataDictType, NULL);
        shadow_big_keys = m_zslCreate();
        RedisModule_Log(ctx, "notice", "start inspector scan task");
    }

    // 当识别到的前缀太多时, 停止继续扫描, 防止占有内存过大
    if (dictSize(shadow_sys_data) > 2000) {
        task_info.is_running = 0;
        long long now = RedisModule_Milliseconds();
        task_info.cost_time = now - task_info.start_time;
        m_dictRelease(task_info.sys_data);
        task_info.sys_data = shadow_sys_data;
        task_info.global_big_key = shadow_big_keys;
        RedisModule_Log(ctx, "warning", "recognize too much system, stop this scan");
        goto restart;
    }

    static long long cursor = 0;
    RedisModuleCallReply *scan_reply = RedisModule_Call(ctx, "SCAN", "lcl", cursor, "COUNT", task_config.scan_count);
    if (scan_reply == NULL) {
        RedisModule_Log(ctx, "warning", "scan reply is null");
        goto restart;
    }

    Module_Assert(RedisModule_CallReplyType(scan_reply) == REDISMODULE_REPLY_ARRAY);
    Module_Assert(RedisModule_CallReplyLength(scan_reply) == 2);

    RedisModuleCallReply *cursor_reply = RedisModule_CallReplyArrayElement(scan_reply, 0);
    Module_Assert(RedisModule_CallReplyType(cursor_reply) == REDISMODULE_REPLY_STRING);
    Module_Assert(RedisModule_StringToLongLong(RedisModule_CreateStringFromCallReply(cursor_reply), &cursor)
                  == REDISMODULE_OK);

    RedisModuleCallReply *keys_reply = RedisModule_CallReplyArrayElement(scan_reply, 1);
    Module_Assert(RedisModule_CallReplyType(keys_reply) == REDISMODULE_REPLY_ARRAY);
    size_t key_num = RedisModule_CallReplyLength(keys_reply);
    for (int i = 0; i < key_num; i++) {
        RedisModuleCallReply *key_reply = RedisModule_CallReplyArrayElement(keys_reply, i);
        Module_Assert(RedisModule_CallReplyType(key_reply) == REDISMODULE_REPLY_STRING);
        RedisModuleString *key = RedisModule_CreateStringFromCallReply(key_reply);

        // 累加扫描到的key个数
        task_info.count += 1;

        if (skip_internal_key(key)) {
            continue;
        }

        RedisModuleKey *real_key = RedisModule_OpenKey(ctx, key, REDISMODULE_READ | REDISMODULE_OPEN_KEY_NOTOUCH);

        uint64_t mem = 0;
        int key_type = 0;
        int is_big = is_big_key(ctx, key, real_key, &mem, &key_type);
        int is_ttl = is_ttl_key(real_key);

        // 在检测big_key或ttl_key过程中, key失效
        if (is_big == -1 || is_ttl == -1) {
            continue;
        }

        SystemUsage *sys_usage;
        RedisModuleString *system_id = extract_sys_id(ctx, key);
        m_dictEntry *entry = m_dictFind(shadow_sys_data, system_id);
        if (entry == NULL) {
            sys_usage = (SystemUsage *)zmalloc(sizeof(SystemUsage));
            memset(sys_usage, 0x00, sizeof(SystemUsage));
            sys_usage->big_key_sample = m_zslCreate();
            sys_usage->ttl_key_sample = m_listCreate();
            listSetFreeMethod(sys_usage->ttl_key_sample, listFreeElement);
            m_dictAdd(shadow_sys_data, system_id, sys_usage);
            RedisModule_RetainString(NULL, system_id);
        } else {
            sys_usage = (SystemUsage *)dictGetVal(entry);
        }

        // 更新对应子系统信息
        sys_usage->capacity += mem;
        sys_usage->key_cnt += 1;

        // 统计子系统级别的key类型
        switch (key_type) {
            case REDISMODULE_KEYTYPE_STRING: sys_usage->key_cnt_string += 1; break;
            case REDISMODULE_KEYTYPE_LIST: sys_usage->key_cnt_list += 1; break;
            case REDISMODULE_KEYTYPE_HASH: sys_usage->key_cnt_hash += 1; break;
            case REDISMODULE_KEYTYPE_SET: sys_usage->key_cnt_set += 1; break;
            case REDISMODULE_KEYTYPE_ZSET: sys_usage->key_cnt_zset += 1; break;
            case REDISMODULE_KEYTYPE_MODULE_TAIR_STRING: sys_usage->key_cnt_tair_string += 1; break;
            case REDISMODULE_KEYTYPE_MODULE_TAIR_HASH: sys_usage->key_cnt_tair_hash += 1; break;
            default: sys_usage->key_cnt_other += 1;
        }

        if (is_big == 1) {
            sys_usage->big_key_cnt += 1;
            sys_usage->big_key_capacity += mem;

            m_zskiplistNode *min_node = m_zslGetElementByRank(sys_usage->big_key_sample, 1);
            if (min_node == NULL || mem >= min_node->score) {
                m_zslInsert(sys_usage->big_key_sample, mem, key);
                RedisModule_RetainString(NULL, key);
                if (sys_usage->big_key_sample->length > SAMPLE_KEY_SIZE) {
                    m_zslDeleteRangeByRank(sys_usage->big_key_sample, 0,
                                           sys_usage->big_key_sample->length - SAMPLE_KEY_SIZE);
                }
            }

            // 记录全局最大big_key
            min_node = m_zslGetElementByRank(shadow_big_keys, 1);
            if (min_node == NULL || mem >= min_node->score) {
                RedisModuleString *g_big_key = RedisModule_CreateStringFromString(ctx, key);
                m_zslInsert(shadow_big_keys, mem, g_big_key);
                RedisModule_RetainString(NULL, g_big_key);
                if (shadow_big_keys->length > SAMPLE_KEY_SIZE) {
                    m_zslDeleteRangeByRank(shadow_big_keys, 0, shadow_big_keys->length - SAMPLE_KEY_SIZE);
                }
            }
        }
        if (is_ttl == 1) {
            sys_usage->ttl_key_cnt += 1;
            sys_usage->ttl_key_capacity += mem;
            if (listLength(sys_usage->ttl_key_sample) < SAMPLE_KEY_SIZE) {
                m_listAddNodeTail(sys_usage->ttl_key_sample, key);
                RedisModule_RetainString(NULL, key);
            }
        }

        RedisModule_CloseKey(real_key);
    }

    // 返回游标为0, 表示结束一轮scan
    if (cursor == 0) {
        task_info.is_running = 0;
        long long now = RedisModule_Milliseconds();
        task_info.cost_time = now - task_info.start_time;
        m_dictRelease(task_info.sys_data);
        task_info.sys_data = shadow_sys_data;
        shadow_sys_data = NULL;
        m_zslFree(task_info.global_big_key);
        task_info.global_big_key = shadow_big_keys;
        shadow_big_keys = NULL;
        RedisModule_Log(ctx, "notice", "finish inspector scan task");
    }

restart:
    task_info.timer_id = RedisModule_CreateTimer(ctx, task_config.scan_tick, inspectorTimerHandler, data);
}

void startInspectorTimer(RedisModuleCtx *ctx, void *data) {
    // 检查是否存在定时任务, 若存在则返回OK, 防止重复创建任务
    if (RedisModule_GetTimerInfo(ctx, task_info.timer_id, NULL, NULL) == REDISMODULE_OK) {
        RedisModule_Log(ctx, "warning", "exist inspector task, should not be invoked");
        return;
    }

    // 首次在module加载后1分钟执行, 后续按配置间隔执行
    task_info.timer_id = RedisModule_CreateTimer(ctx, 60 * 1000, inspectorTimerHandler, data);
    RedisModule_Log(ctx, "notice", "create inspector timer");
}

int __attribute__((visibility("default"))) RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    if (RedisModule_Init(ctx, "inspector", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR) {
        return REDISMODULE_ERR;
    }

    if (RedisModule_CreateCommand(ctx, "inspector.info", InspectorInfo_RedisCommand, "readonly", 0, 0, 0)
        != REDISMODULE_OK) {
        RedisModule_Log(ctx, "warning", "register inspector info cmd error");
        return REDISMODULE_ERR;
    }
    if (RedisModule_CreateCommand(ctx, "inspector.keys", InspectorKeys_RedisCommand, "readonly", 0, 0, 0)
        != REDISMODULE_OK) {
        RedisModule_Log(ctx, "warning", "register inspector keys cmd error");
        return REDISMODULE_ERR;
    }
    if (RedisModule_CreateCommand(ctx, "inspector.config", InspectorConfig_RedisCommand, "readonly", 0, 0, 0)
        != REDISMODULE_OK) {
        RedisModule_Log(ctx, "warning", "register inspector config cmd error");
        return REDISMODULE_ERR;
    }
    if (RedisModule_CreateCommand(ctx, "inspector.trigger", InspectorTrigger_RedisCommand, "readonly", 0, 0, 0)
        != REDISMODULE_OK) {
        RedisModule_Log(ctx, "warning", "register inspector trigger cmd error");
        return REDISMODULE_ERR;
    }

    // 初始化默认配置
    task_config.enable = 0;
    task_config.exclusive = 0;
    task_config.interval = DEFAULT_INSPECTOR_TIMER_INTERVAL;
    task_config.scan_tick = DEFAULT_INSPECTOR_SCAN_TICK;
    task_config.scan_count = DEFAULT_INSPECTOR_SCAN_COUNT;
    task_config.items_threshold = DEFAULT_INSPECTOR_ITEMS_THRESHOLD;
    task_config.memory_threshold = DEFAULT_INSPECTOR_MEMORY_THRESHOLD;
    task_config.ttl_threshold = DEFAULT_INSPECTOR_TTL_THRESHOLD;

    if (argc > 0) {
        config_path = RedisModule_StringPtrLen(argv[0], NULL);
    }
    if (config_path != NULL && load_config(ctx, config_path, &task_config) != 0) {
        return REDISMODULE_ERR;
    }

    memset(&task_info, 0x00, sizeof(TaskInfo));
    task_info.sys_data = m_dictCreate(&inspectDataDictType, NULL);
    task_info.global_big_key = m_zslCreate();

    // 创建定时任务
    startInspectorTimer(ctx, NULL);

    return REDISMODULE_OK;
}

int __attribute__((visibility("default"))) RedisModule_OnUnload(RedisModuleCtx *ctx) {
    if (RedisModule_StopTimer(ctx, task_info.timer_id, NULL) == REDISMODULE_ERR) {
        RedisModule_Log(ctx, "warning", "stop inspector timer error");
    }

    task_info.start_time = 0;
    task_info.cost_time = 0;
    task_info.count = 0;
    task_info.dbsize = 0;
    task_info.timer_id = 0;
    if (task_info.sys_data) m_dictRelease(task_info.sys_data);
    if (task_info.global_big_key) m_zslFree(task_info.global_big_key);
    if (shadow_sys_data) {
        m_dictRelease(shadow_sys_data);
        shadow_sys_data = NULL;
    }
    if (shadow_big_keys) {
        m_zslFree(shadow_big_keys);
        shadow_big_keys = NULL;
    }

    return REDISMODULE_OK;
}